/*
 * Copyright (c) 2024 Black Sesame Technologies
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include "dal_fastdds/entity_impl/DalDataReaderImpl.hpp"
#include "dal_fastdds/utils/log.hpp"
#include "dal_fastrtps_cpp/type_support_common.hpp"

DalDataReaderImpl::DalDataReaderImpl(DataReader* fastdds_datareader,
                                     DataReaderListenerImpl* listener)
{
    fastdds_datareader_ = fastdds_datareader;
    listener_ = listener;
}

DalDataReaderImpl::~DalDataReaderImpl()
{
#ifdef DAL_ENABLE_DMABUFHEAP
    std::lock_guard<std::mutex> lock(mutex_);
    if (dbh_reader_)
        delete dbh_reader_;
#endif
}

DataReader*& DalDataReaderImpl::FastddsDataReader()
{
    return fastdds_datareader_;
}

void DalDataReaderImpl::SetFastddsDataReader(DataReader* fastdds_datareader)
{
    fastdds_datareader_ = fastdds_datareader;
}

Subscriber*& DalDataReaderImpl::FastddsSubscriber()
{
    return fastdds_sub_;
}

void DalDataReaderImpl::SetFastddsSubscriber(Subscriber*& subscriber)
{
    fastdds_sub_ = subscriber;
}

DomainParticipant*& DalDataReaderImpl::FastddsDomainParticipant()
{
    return fastdds_participant_;
}

void DalDataReaderImpl::SetFastddsDomainParticipant(DomainParticipant*& participant)
{
    fastdds_participant_ = participant;
}

void DalDataReaderImpl::SetTypeSupport(const void* type_support)
{
    type_support_ = type_support;
}

ReturnCode_t DalDataReaderImpl::DDSTake(void* data, SampleInfoSeq& info)
{
    std::lock_guard<std::mutex> lock(mutex_);

    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = data;
    sdata.impl = type_support_;

    StackAllocatedSequence<void*, 1> data_values;
    const_cast<void**>(data_values.buffer())[0] = &sdata;

    return fastdds_datareader_->take(data_values, info, 1);
}

ReturnCode_t DalDataReaderImpl::DDSRead(void* data, SampleInfoSeq& info)
{
    std::lock_guard<std::mutex> lock(mutex_);

    dal_fastrtps_shared_cpp::SerializedData sdata;
    sdata.is_cdr_buffer = false;
    sdata.data = data;
    sdata.impl = type_support_;

    StackAllocatedSequence<void*, 1> data_values;
    const_cast<void**>(data_values.buffer())[0] = &sdata;

    return fastdds_datareader_->read(data_values, info, 1);
}

#ifdef DAL_ENABLE_DMABUFHEAP
ReturnCode_t DalDataReaderImpl::DBHTake(void* data, SampleInfoSeq& info)
{
    std::lock_guard<std::mutex> lock(mutex_);

    if (!dbh_reader_->RecvDbhSendNotifyMsg(info)) {
        LOG_ERROR_WITH_LOCATION() << "RecvDbhSendNotifyMsg failed" << std::endl;
        return ReturnCode_t::RETCODE_ERROR;
    }

    dbh_reader_->ParseDbhSendNotifyMsg();
    dbh_reader_->SetDbhRecvNotifyMsg();
    if (!dbh_reader_->GetMsgFromDbhBlock(data)) {
        LOG_ERROR_WITH_LOCATION() << "GetMsgFromDbhBlock failed" << std::endl;
        return ReturnCode_t::RETCODE_ERROR;
    }

    if (!dbh_reader_->SendDbhRecvNotifyMsg()) {
        LOG_ERROR_WITH_LOCATION() << "SendDbhRecvNotifyMsg failed" << std::endl;
        return ReturnCode_t::RETCODE_ERROR;
    }

    return ReturnCode_t::RETCODE_OK;
}
#endif
